package org.iscas.tcse.util;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Collections;
import java.util.Properties;
import java.util.Set;

public class KafkaUtil {
    public static void prepareTopic(String addr, String topic, int partition) throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", addr);
        AdminClient adminClient = AdminClient.create(props);
        Set<String> topics = adminClient.listTopics().names().get();
        if (topics.contains(topic)) {
            adminClient.deleteTopics(Collections.singletonList(topic));
        }
        Thread.sleep(2000);
        adminClient.createTopics(
                Collections.singletonList(new NewTopic(topic, partition, (short) 1)));
        adminClient.close();
    }
}
